//Team 4 - Gold Coast
//Duy Kha Nguyen 598240 
//Rongkai Zhou 562946 
//Zhangyue Cheng 548692
//Binzan Liang 591911
//Wei Liu 572322
package couchdb;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

import org.apache.log4j.Logger;
import org.lightcouch.CouchDbClient;

public class MergeDbApp {

    private static final Logger logger = Logger.getLogger(MergeDbApp.class);

    public static void main(String[] args) {
        final BlockingQueue<Map> tweetQueue = new ArrayBlockingQueue<Map>(5000);
        final String[] dbHosts = { "115.146.95.53", "115.146.86.148", "115.146.95.28", "115.146.95.40" };

        final List<Thread> exportThreads = new ArrayList<Thread>();
        for (final String host : dbHosts) {
            final CouchDbClient exportSource = new CouchDbClient("db-tweet", false, "http", host, 5984, null, null);
            exportThreads.add(new Thread(new ExportTweetTask(exportSource, tweetQueue)));
        }

        final CouchDbClient importSource = new CouchDbClient("db-tweet-final", true, "http", "localhost", 5984, null, null);
        final List<ImportTweetTask> importTasks = new ArrayList<ImportTweetTask>();
        final List<Thread> importThreads = new ArrayList<Thread>();
        for (int i = 0; i < 4; i++) {
            final ImportTweetTask importTask = new ImportTweetTask(importSource, tweetQueue);

            importTasks.add(importTask);
            importThreads.add(new Thread(importTask));
        }

        for (final Thread exportThread : exportThreads) {
            exportThread.start();
        }
        for (final Thread importThread : importThreads) {
            importThread.start();
        }

        try {
            for (final Thread exportThread : exportThreads) {
                exportThread.join();
            }

            for (final ImportTweetTask importTask : importTasks) {
                importTask.stop();
            }

            for (final Thread importThread : importThreads) {
                importThread.join();
            }
        } catch (Exception e) {
            logger.error("Unexpected exception", e);
        }

    }
}
